package producer;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

/**
 * @author 夏焰波
 * @date 2020-09-13 21:08
 */
public class CallBackProducer {
    public static void main(String[] args) {
        //1.创建配置信息
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        //2. 创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        /**
         回调函数会在
         producer收到 ack时调用，为异步调用， 该方法有两个参数，分别是
         RecordMetadata 和 Exception ，如果 Exception 为 null ，说明消息发送成功，如果
         Exception 不为 null ，说明消息发送失败。
         注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。
         */
        //3.发送数据
        for (int i = 0; i < 10; i++) {
            //producer.send(new ProducerRecord<String, String>("first", "value--" + Integer.toString(i)),
            producer.send(new ProducerRecord<String, String>("first", 0, "key" + i, "value--" + Integer.toString(i)),
                    new Callback() {
                        //回调函数， 该方法会在 Producer 收到 ack 时调用，为异步调用
                        @Override
                        public void onCompletion(RecordMetadata metadata, Exception exception) {
                            if (exception == null) {
                                //offset从0开始
                                System.out.println(metadata.partition() + " -- " + metadata.offset());
                            } else {
                                exception.printStackTrace();
                            }
                        }
                    });

            //4.关闭资源，会清缓存
            producer.close();
        }
    }
}
